package com.cbh.amadeus.server.handler;

import com.cbh.amadeus.common.enums.MessageApproach;
import com.cbh.amadeus.common.enums.MessageType;
import com.cbh.amadeus.common.protocol.AmadeusMessage;
import com.cbh.amadeus.common.protocol.AmadeusMessageWrap;
import com.cbh.amadeus.server.domain.ChannelModel;
import com.cbh.amadeus.server.utils.MessageApproachManage;
import com.cbh.amadeus.server.utils.MessageRejectManage;
import com.cbh.amadeus.server.utils.TopicChannelManage;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import lombok.extern.slf4j.Slf4j;

import java.util.List;

/**
 * @author ChenBiHui
 * @date 2025/4/25
 */
@Slf4j
public class PublishHandler extends SimpleChannelInboundHandler<AmadeusMessageWrap> {

    public static void publish(Channel client, AmadeusMessageWrap msg) {
        AmadeusMessage message = msg.getMessage();
        String topic = message.getTopic();
        String messageId = message.getMessageId();
        List<ChannelModel> models = TopicChannelManage.findModelWithTopic(topic);
        if (models == null || models.isEmpty()) {
            client.writeAndFlush(MessageRejectManage.reject(msg))
                    .addListener(item -> {
                        if (item.isSuccess()) {
                            log.warn(
                                    "===== [GEN REJECT] Topic [{}] message-id [{}] can't find any channel, has reject this message =====",
                                    topic,
                                    messageId
                            );
                        }
                    });

            return;
        }
        for (ChannelModel model : models) {
            Channel channel = model.getChannel();
            if (channel.isActive()) {
                channel.writeAndFlush(msg)
                        .addListener(item -> {
                            if (item.isSuccess()) {
                                log.info(
                                        "===== [GEN] Topic [{}] message-id [{}] has publish to the register [{}] =====",
                                        topic,
                                        messageId,
                                        model.getOrigin()
                                );
                            }
                        });
            }
        }
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, AmadeusMessageWrap msg) {
        MessageType type = MessageType.find(msg.getType());
        if (type == MessageType.GEN) {
            MessageApproach approach = MessageApproach.find(msg.getApproach());
            switch (approach) {
                case DIRECT:
                    PublishHandler.publish(ctx.channel(), msg);
                    break;
                case DELAY:
                    MessageApproachManage.push(ctx.channel(), msg);
                    break;
                default:
                    break;
            }
            return;
        }
        ctx.fireChannelRead(msg);
    }
}
